package com.bluedot.queue;

import com.bluedot.message.Message;

import java.util.concurrent.TimeUnit;

/**
 * 程序传递返回结果使用的队列的抽象父类<br/>
 * 定义并且实现了2个子类都会使用到的方法
 */
public abstract class OutQueue extends AbstractQueue {

    protected OutQueue() {
    }

    /**
     * 通过特定唯一的key从队列中取消息,不设置超时时间,当队列为空时当前线程进入等待队列emptyWaitSet进行等待,队列一直为空则一直等待,直到有别的线程唤醒<br/>
     * 从队列中取消息前先获取锁,获取失败则当前线程进入阻塞队列且被interrupt,直到其他线程释放锁当前线程再进行锁竞争<br/>
     * 成功取出消息时要随机唤醒一个在fullWaitSet中等待的线程
     *
     * @return 从队列中取出的消息, 本方法队列为空时会一直等待, 所以一定不返回null值
     */
    public Message poll(String key) {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            Message message = queue.stream().filter(msg -> msg.getKey().equals(key)).findAny().get();
            queue.removeIf(msg -> msg.getKey().equals(key));
            fullWaitSet.signal();
            return message;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 通过特定唯一的key从队列中取消息,设置超时时间,当队列为空时当前线程进入等待队列emptyWaitSet进行等待,若等待时间超过设置的超时时间则放弃取消息,直接返回null值<br/>
     * 从队列中取消息前先获取锁,获取失败则当前线程进入阻塞队列且被interrupt,直到其他线程释放锁当前线程再进行锁竞争<br/>
     * 成功取出消息时要随机唤醒一个在fullWaitSet中等待的线程
     *
     * @param timeout  超时时间
     * @param timeUnit 超时时间单位
     * @return 从队列中取出的消息
     */
    public Message poll(String key, long timeout, TimeUnit timeUnit) {
        lock.lock();
        try {
            long nanos = timeUnit.toNanos(timeout);
            while (queue.isEmpty()) {
                try {
                    if (nanos <= 0) {
                        return null;
                    }
                    nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            Message message = queue.stream().filter(msg -> msg.getKey().equals(key)).findAny().get();
            queue.removeIf(msg -> msg.getKey().equals(key));
            fullWaitSet.signal();
            return message;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 获取当前队列头的消息但不取出
     *
     * @return 当前队列头的消息
     */
    public Message peek() {
        lock.lock();
        try {
            return queue.peekFirst();
        } finally {
            lock.unlock();
        }
    }
}
